package org.ala.tcp_feign_client.context;

import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

import org.ala.tcp_feign_client.ITcpFeignClient;
import org.ala.tcp_feign_client.context.heart.HeartScheduled;
import org.ala.tcp_feign_client.context.reconnect.ReconnectScheduled;
import org.ala.tcp_feign_client.impl.DefaultTcpFeignClient;
import org.ala.tcp_feign_common.config.AttrbuteKeyNamespace;
import org.ala.tcp_feign_common.config.TcpFeignConfigure;
import org.ala.tcp_feign_common.logs.LoggerFactory;
import org.ala.tcp_feign_common.serialize.ISerializabler;
import org.ala.tcp_feign_common.serialize.kryo.KryoSerializabler;
import org.slf4j.Logger;

import io.netty.channel.Channel;
import io.netty.util.AttributeKey;


/**
 * tcp-feign 客户端context
 *
 * @author 骆毅(ala)
 * @date 2024年4月20日
 */
public class TcpFeignClientContext {

	
	static Logger log = LoggerFactory.getLogger("tcp-feign", "client");
	
	
	/**
	 * 共享客户端后缀
	 */
	public static final String SUFFIX_COMMON = "COMMON";
	
	
	/**
	 * 单利
	 */
	static TcpFeignClientContext _instance;
	public static TcpFeignClientContext instance() {
		if (_instance == null) {
			synchronized (TcpFeignClientContext.class) {
				if (_instance == null) {
					_instance = new TcpFeignClientContext();
				}
			}
		}
		return _instance;
	}
	
	
	/**
	 * 活跃链路holder
	 * <p>	链接成功，并且登录成功的channel
	 */
	protected Map<String, ITcpFeignClient> activeChannel;
	/**
	 * 心跳检测定时任务
	 */
	protected HeartScheduled heartScheduled;
	/**
	 * 重连定时任务
	 */
	protected ReconnectScheduled reconnectScheduled;
	/**
	 * 序列化/反序列化工具
	 */
	protected ISerializabler serializabler;
	/**
	 * resp线程池
	 */
	protected ThreadPoolExecutor clientBizRespThreadPool;
	/**
	 * 初始化
	 */
	public void init(TcpFeignConfigure config) {
		activeChannel = new ConcurrentHashMap<>();
		
		heartScheduled = new HeartScheduled(config.getHeartInterval());
		heartScheduled.start();
		
		reconnectScheduled = new ReconnectScheduled();
		reconnectScheduled.start();
		
		serializabler = new KryoSerializabler();
		
		clientBizRespThreadPool = new ThreadPoolExecutor(config.getClientRespBizThreadPoolCoreSize(), config.getClientRespBizThreadPoolMaxSize(),
											180, TimeUnit.SECONDS,
											new ArrayBlockingQueue<>(config.getClientRespBizThreadPoolQueueSize()),
											new ThreadFactory() {
												AtomicInteger count = new AtomicInteger(0);
												public Thread newThread(Runnable r) {
													Thread t = new Thread(r);
													t.setName("ClientBizRespThread_" + count.getAndIncrement());
													return t;
												}
											});
	}
	
	
	/**
	 * 建立链接
	 * @throws InterruptedException 
	 */
	public ITcpFeignClient getOrConnectClient(String host, int port, String keySuffix, String username, String password, Consumer<ITcpFeignClient> loginSuccess) throws Exception {
		ITcpFeignClient client = getOrCreateClient(host, port, keySuffix, username, password, loginSuccess);
		client.connect();
		return client;
	}
	/**
	 * 获取或者建立链接
	 */
	public ITcpFeignClient getOrCreateClient(String host, int port, String keySuffix, String username, String password, Consumer<ITcpFeignClient> loginSuccess) {
		String key = key(host, port, keySuffix, username, password);
		ITcpFeignClient client = activeChannel.get(key);
		if (client == null) {
			synchronized (key.intern()) {
				if (client == null) {
					client = new DefaultTcpFeignClient(host, port, keySuffix, username, password);
					if (loginSuccess != null) {client.bindLoginCallback(loginSuccess);}
					activeChannel.put(key, client);
				}
			}
		}
		return client;
	}
	/**
	 * 获取链接
	 */
	public ITcpFeignClient getClient(String host, int port, String keySuffix, String username, String password) {
		return activeChannel.get(key(host, port, keySuffix, username, password));
	}
	
	
	/**
	 * 注册活跃链路
	 */
	public void register(ITcpFeignClient client) {
	}
	/**
	 * host + port组成的key
	 */
	protected String key(String host, int port, String keySuffix, String username, String password) {
		return host + ":" + port + ":" + keySuffix + ":" + username + ":" + password;
	}
	
	
	/**
	 * 关闭连接
	 */
	public void close(String host, int port, String keySuffix, String username, String password) {
		String key = key(host, port, keySuffix, username, password);
		ITcpFeignClient client = activeChannel.get(key);
		if (client != null) {
			client.close();
			activeChannel.remove(key);
		}
	}
	
	
	public Map<String, ITcpFeignClient> getActiveChannel() {
		return activeChannel;
	}
	
	
	public ReconnectScheduled getReconnectScheduled() {
		return reconnectScheduled;
	}


	public ISerializabler getSerializabler() {
		return serializabler;
	}
	
	
	public ThreadPoolExecutor getClientBizRespThreadPool() {
		return clientBizRespThreadPool;
	}


	/**
	 * 工具方法，通过channel取host
	 */
	public static String hostFromChannel(Channel channel) {
		return (String)channel.attr(AttributeKey.valueOf(AttrbuteKeyNamespace.REMOTE_HOST)).get();
	}
	/**
	 * 工具方法，通过channel取port
	 */
	public static Integer portFromChannel(Channel channel) {
		return (Integer)channel.attr(AttributeKey.valueOf(AttrbuteKeyNamespace.REMOTE_PORT)).get();
	}
	/**
	 * 工具方法，通过channel取suffix
	 */
	public static String suffixFromChannel(Channel channel) {
		return (String)channel.attr(AttributeKey.valueOf(AttrbuteKeyNamespace.CLIENT_SUFFIX)).get();
	}
	/**
	 * 工具方法，通过channel取username
	 */
	public static String userFromChannel(Channel channel) {
		return (String)channel.attr(AttributeKey.valueOf(AttrbuteKeyNamespace.CLIENT_USERNAME)).get();
	}
	/**
	 * 工具方法，通过channel取password
	 */
	public static String passwordFromChannel(Channel channel) {
		return (String)channel.attr(AttributeKey.valueOf(AttrbuteKeyNamespace.CLIENT_PASSWORD)).get();
	}
}
